package com.wuwei.elearning.messageQueen;

import com.wuwei.elearning.dto.ConversionMessage;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * In-memory implementation of a message queue
 */
public class MessageQueue implements MessageProducer {
    private final Map<String, PriorityBlockingQueue<ConversionMessage>> topicQueues;
    private final Map<String, List<MessageConsumer>> topicConsumers;
    private final Map<Long, ConversionMessage> messageStore;
    private final PriorityBlockingQueue<ScheduledMessage> scheduledMessages;
    private final ExecutorService executorService;
    private final ScheduledExecutorService schedulerService;
    private final AtomicBoolean isRunning;
    private final int maxRetries;
    private final long retryDelayMs;

    private static class ScheduledMessage implements Comparable<ScheduledMessage> {
        private final ConversionMessage message;
        private final LocalDateTime scheduledTime;

        public ScheduledMessage(ConversionMessage message, LocalDateTime scheduledTime) {
            this.message = message;
            this.scheduledTime = scheduledTime;
        }

        public ConversionMessage getMessage() {
            return message;
        }

        public LocalDateTime getScheduledTime() {
            return scheduledTime;
        }

        @Override
        public int compareTo(ScheduledMessage other) {
            return this.scheduledTime.compareTo(other.scheduledTime);
        }
    }

    public MessageQueue(int numThreads, int maxRetries, long retryDelayMs) {
        this.topicQueues = new ConcurrentHashMap<>();
        this.topicConsumers = new ConcurrentHashMap<>();
        this.messageStore = new ConcurrentHashMap<>();
        this.scheduledMessages = new PriorityBlockingQueue<>();
        this.executorService = Executors.newFixedThreadPool(numThreads);
        this.schedulerService = Executors.newScheduledThreadPool(1);
        this.isRunning = new AtomicBoolean(false);
        this.maxRetries = maxRetries;
        this.retryDelayMs = retryDelayMs;
    }

    public MessageQueue() {
        this(Runtime.getRuntime().availableProcessors(), 3, 5000);
    }

    /**
     * Start the message queue processing
     */
    public void start() {
        if (isRunning.compareAndSet(false, true)) {
            // Start the message processing
            for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
                executorService.submit(this::processMessages);
            }

            // Start the scheduler for delayed messages
            schedulerService.scheduleAtFixedRate(this::processScheduledMessages, 0, 1, TimeUnit.SECONDS);

            System.out.println("ConversionMessage queue started");
        }
    }

    /**
     * Stop the message queue processing
     */
    public void stop() {
        if (isRunning.compareAndSet(true, false)) {
            executorService.shutdown();
            schedulerService.shutdown();
            try {
                if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
                if (!schedulerService.awaitTermination(10, TimeUnit.SECONDS)) {
                    schedulerService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                schedulerService.shutdownNow();
                Thread.currentThread().interrupt();
            }
            System.out.println("ConversionMessage queue stopped");
        }
    }

    /**
     * Register a consumer for a specific topic
     *
     * @param consumer The consumer to register
     */
    public void registerConsumer(MessageConsumer consumer) {
        String topic = consumer.getTopic();
        topicConsumers.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>()).add(consumer);
        topicQueues.computeIfAbsent(topic, k -> new PriorityBlockingQueue<>(100,
                Comparator.comparing(ConversionMessage::getPriority).reversed()
                          .thenComparing(ConversionMessage::getCreatedAt)));
        System.out.println("Registered consumer for topic: " + topic);
    }

    /**
     * Unregister a consumer for a specific topic
     *
     * @param consumer The consumer to unregister
     */
    public void unregisterConsumer(MessageConsumer consumer) {
        String topic = consumer.getTopic();
        List<MessageConsumer> consumers = topicConsumers.get(topic);
        if (consumers != null) {
            consumers.remove(consumer);
            if (consumers.isEmpty()) {
                topicConsumers.remove(topic);
            }
        }
        System.out.println("Unregistered consumer for topic: " + topic);
    }

    @Override
    public ConversionMessage send(String topic, Object payload,Long id) {
        return send(topic, payload, 0,id);
    }

    @Override
    public ConversionMessage send(String topic, Object payload, int priority,Long id) {
        ConversionMessage message = new ConversionMessage(topic, payload, priority,id);
        enqueueMessage(message);
        return message;
    }

    @Override
    public ConversionMessage schedule(String topic, Object payload, LocalDateTime scheduledTime) {
        ConversionMessage message = new ConversionMessage(topic, payload, scheduledTime);
        message.setStatus(ConversionMessage.MessageStatus.PENDING);
        messageStore.put(message.getId(), message);
        scheduledMessages.offer(new ScheduledMessage(message, scheduledTime));
        return message;
    }

    /**
     * Get a message by its ID
     *
     * @param messageId The message ID
     * @return The message, or null if not found
     */
    public ConversionMessage getMessage(String messageId) {
        return messageStore.get(messageId);
    }

    /**
     * Get all messages for a specific topic
     *
     * @param topic The topic
     * @return A list of messages
     */
    public List<ConversionMessage> getMessagesByTopic(String topic) {
        return messageStore.values().stream()
                .filter(message -> message.getTopic().equals(topic))
                .collect(Collectors.toList());
    }

    /**
     * Get all messages with a specific status
     *
     * @param status The status
     * @return A list of messages
     */
    public List<ConversionMessage> getMessagesByStatus(ConversionMessage.MessageStatus status) {
        return messageStore.values().stream()
                .filter(message -> message.getStatus() == status)
                .collect(Collectors.toList());
    }

    /**
     * Add a message to the queue
     *
     * @param message The message to add
     */
    private void enqueueMessage(ConversionMessage message) {
        messageStore.put(message.getId(), message);
        topicQueues.computeIfAbsent(message.getTopic(), k -> new PriorityBlockingQueue<>(100,
                Comparator.comparing(ConversionMessage::getPriority).reversed()
                          .thenComparing(ConversionMessage::getCreatedAt)))
                .offer(message);
    }

    /**
     * Process messages from the queue
     */
    private void processMessages() {
        while (isRunning.get()) {
            try {
                // Find a topic with messages
                for (Map.Entry<String, PriorityBlockingQueue<ConversionMessage>> entry : topicQueues.entrySet()) {
                    String topic = entry.getKey();
                    PriorityBlockingQueue<ConversionMessage> queue = entry.getValue();

                    ConversionMessage message = queue.poll();
                    if (message != null) {
                        processMessage(message);
                    }
                }

                // Sleep a bit to avoid busy waiting
                Thread.sleep(10);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                System.err.println("Error processing messages: " + e.getMessage());
                e.printStackTrace();
            }
        }
    }

    /**
     * Process scheduled messages that are due
     */
    private void processScheduledMessages() {
        LocalDateTime now = LocalDateTime.now();
        List<ScheduledMessage> dueMessages = new ArrayList<>();

        // Find messages that are due
        while (!scheduledMessages.isEmpty() &&
               scheduledMessages.peek().getScheduledTime().isBefore(now)) {
            ScheduledMessage scheduledMessage = scheduledMessages.poll();
            if (scheduledMessage != null) {
                dueMessages.add(scheduledMessage);
            }
        }

        // Process due messages
        for (ScheduledMessage scheduledMessage : dueMessages) {
            ConversionMessage message = scheduledMessage.getMessage();
            enqueueMessage(message);
        }
    }

    /**
     * Process a single message
     *
     * @param message The message to process
     */
    private void processMessage(ConversionMessage message) {
        List<MessageConsumer> consumers = topicConsumers.get(message.getTopic());
        if (consumers == null || consumers.isEmpty()) {
            // No consumers for this topic, mark as failed
            message.setStatus(ConversionMessage.MessageStatus.FAILED);
            return;
        }

        message.setStatus(ConversionMessage.MessageStatus.PROCESSING);

        boolean success = false;
        for (MessageConsumer consumer : consumers) {
            try {
                success = consumer.consume(message);
                if (success) {
                    break;
                }
            } catch (Exception e) {
                System.err.println("Error processing message: " + e.getMessage());
                e.printStackTrace();
            }
        }

        if (success) {
            message.setStatus(ConversionMessage.MessageStatus.COMPLETED);
        } else {
            message.incrementRetryCount();
            if (message.getRetryCount() < maxRetries) {
                // Schedule for retry
                message.setStatus(ConversionMessage.MessageStatus.PENDING);
                schedulerService.schedule(() -> enqueueMessage(message),
                        retryDelayMs * (1L << message.getRetryCount()), TimeUnit.MILLISECONDS);
            } else {
                message.setStatus(ConversionMessage.MessageStatus.FAILED);
            }
        }
    }
} 